package com.gitee.cui.basic.producerandconsumer;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.gitee.cui.petstore.actor.Consumer;
import com.gitee.cui.petstore.actor.Producer;
import com.gitee.cui.petstore.goods.Goods;
import com.gitee.cui.petstore.goods.IGoods;
import com.gitee.cui.util.JvmUtil;
import com.gitee.cui.util.Print;

/**
 * 线程通知机制的安全处理方式
 *
 * @author cuiqingqiang
 * @version 1.0
 * @date 2021-11-20 15:29
 **/
public class CommunicatePetStore {

	private static final int MAX_AMOUNT = 10;

	static class DataBuffer<T> {

		private List<T> dataList = new LinkedList<>();

		private int amount = 0;

		private final Object LOCK = new Object();

		private final Object NOT_FULL = new Object();

		private final Object NOT_EMPTY = new Object();

		public void add(T element) throws InterruptedException {
			while (amount > MAX_AMOUNT) {
				synchronized (NOT_FULL) {
					Print.tcfo("队列已经满了");
					NOT_FULL.wait();
				}
			}

			synchronized (LOCK) {
				dataList.add(element);
				amount++;
			}

			// 添加进一个数据，就通知取消费者线程，缓冲区有数据，可以获取
			synchronized (NOT_EMPTY) {
				NOT_EMPTY.notify();
			}
		}

		public T fetch() throws InterruptedException {
			while (amount <= 0) {
				synchronized (NOT_EMPTY) {
					Print.tcfo("队列已经空了");
					NOT_EMPTY.wait();
				}
			}

			T element = null;
			synchronized (LOCK) {
				element = dataList.remove(0);
				amount--;
			}

			// 取到一个数据，就通知生产者线程没有满，可以继续添加数据
			synchronized (NOT_FULL) {
				NOT_FULL.notify();
			}

			return element;
		}
	}

	public static void main(String[] args) {
		Print.cfo("当前进程的id是：" + JvmUtil.getProcessID());
		System.setErr(System.out);

		DataBuffer<IGoods> dataBuffer = new DataBuffer<>();

		Callable<IGoods> producerAction = () -> {
			// 随机生成一个商品
			IGoods iGoods = Goods.produceOne();
			// 添加入缓冲区
			dataBuffer.add(iGoods);
			return iGoods;
		};

		Callable<IGoods> consumerAction = () -> {
			// 从缓冲区中获取一个数据
			IGoods goods = dataBuffer.fetch();
			return goods;
		};

		// 创建一个20个线程的线程池
		ExecutorService executorService = Executors.newFixedThreadPool(20);

		// 假设11个线程，10个消费者，一个生产者
		int producerThread = 1;
		int consumerThread = 10;

		for (int i = 0; i < producerThread; i++) {
			executorService.submit(new Producer(producerAction, 50));
		}

		for (int i = 0; i < consumerThread; i++) {
			executorService.submit(new Consumer(consumerAction, 100));
		}
	}
}
